package com.hujiang.basic.search.handler;

import com.hujiang.basic.search.handler.impl.model.HandlerInfo;
import com.hujiang.basic.search.model.HJResponse;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
@Api("增量索引MQ api")
@RequestMapping(value = "/v1/listener")
public class HandlerController {

    @Autowired
    private HandlerService service;

    @RequestMapping(method = RequestMethod.POST)
    @ApiOperation(value = "创建监听器", httpMethod = "POST", response = HJResponse.class, notes = "queue的vhost必须为search")
    public HJResponse addHandler(@RequestParam String queueName, @RequestParam String alias) {
        service.insert(new HandlerInfo(queueName, alias));
        return HJResponse.ok("success");
    }

    @RequestMapping(value = "/pause", method = RequestMethod.POST)
    @ApiOperation(value = "暂停增量更新", httpMethod = "POST", response = HJResponse.class, notes = "queue的vhost必须为search")
    public HJResponse pauseHandler(@RequestParam String queueName) {
        service.pause(queueName);
        return HJResponse.ok("success");
    }

    @RequestMapping(value = "/status", method = RequestMethod.POST)
    @ApiOperation(value = "增量更新", httpMethod = "POST", response = HJResponse.class, notes = "queue的vhost必须为search")
    public HJResponse getListeningQueues(@RequestParam String queueName, @RequestParam String alias) {
        service.insert(new HandlerInfo(queueName, alias));
        service.listen(queueName);
        return HJResponse.ok("success");
    }

    /**
     * 定时扫描queue的状态
     */
    @Scheduled(fixedRate = 60_000)
    public synchronized void registerHandler() {
        List<HandlerInfo> handlers = service.query();
        for (HandlerInfo handlerInfo : handlers) {
            // 添加新的queue
            if (handlerInfo.isActive() && !service.listeningQueues.contains(handlerInfo.getQueue())) {
                service.listen(handlerInfo.getQueue());
                service.listeningQueues.add(handlerInfo.getQueue());
                continue;
            }
            // 暂停一个queue，全量索引时，增量索引关闭。即不再消费queue中消息。直到全量索引结束
            if (!handlerInfo.isActive() && service.listeningQueues.contains(handlerInfo.getQueue())) {
                service.stopListen(handlerInfo.getQueue());
                service.listeningQueues.remove(handlerInfo.getQueue());
            }
        }
    }
}
